/**
 *         
 *                       **  数据采集程序相关说明  **
 *
 *                    程序涉及到modbus,xml,kafka,ntp等内容    
 *
 * 对之前的代码做了一些调整,基本结构没变,修改了一些不合理的地方
 * 增加了判断功能,之前的代码没有对采集到的数据与xml中的阈值数据进行比较来确定发送消息的内容(警告或事件)
 * 或者说之前的代码做了判断但是代码是错误的,之前代码的判断条件永远为真,可能是一个失误
 *
 * if(change(warning_info[f][2],f))   ===> 条件永远为真,导致消息的警告字段与原来xml中是一样的
 * {
 *	sprintf(buf_test,"%s",warning_info[f][0]);	
 *	sprintf(buf_test1,"%s",warning_info[f][1]);	
 *	sprintf(buf_test2,"%s",warning_info[f][3]);
 * }
 *
 * 这份代码不是按照之前定义的判断条件来判断的,那种方法实现起来很复杂,这里直接用采集的值和上下限进行比较,看
 * 是否触发警告和事件,只需对263行的判断条件进行更改就可以了.
 *
 * usage:
 *     make
 *     chmod 777 arm
 *     ./arm *.xml delay(1000000) kafka_broker:ip:9092 realdata_receive collect_receive ntp_server_ip
 *     确保kafka服务开启,指定一个ntp时间服务器来同步时间(建议在arm工控机上事先同步时间,因为ntp同步时间是平
 *     滑的,不会一下子就确保同步)   
 *
 * @MOD : zhujiadong(melo)  
 * @DATE: Thu Dec 22 09:54:25 CST 2016
 *
 **/

#include <stdio.h>
#include <string.h>
#include <libxml/tree.h>
#include <libxml/parser.h>
#include <libxml/xpath.h>
#include <errno.h>
#include <unistd.h>
#include <modbus.h>
#include <sys/time.h>
#include <time.h>

#include "rdkafka.h"

#define SUM_SENSOR 8
#define SUM_SLAVE 2
xmlNodePtr node[80]={0};
int count;   //sensor in use 

char state_info[SUM_SENSOR][5][40]={"0"}; //every sensor's state(from xml) 
char warning_info[SUM_SENSOR][4][80]={"0"}; //warning info
char head[80][2][80]={"0"};        //head is the first line of node 
char *name = NULL;
char *value = NULL;
char dev_basic[SUM_SENSOR][6][80]={"0"};   //dev info in the node 6 = 2 * 3 three lines each line has 2 content 

//to save real warning info
char buf_test[32];
char buf_test1[32];
char buf_test2[32];

char buf_cur[2048];   //实时数据
char buf_his[2048];   //历史数据
char sensor[SUM_SENSOR][60]=
{
    "Channel.Drive.group.vibration1",
    "Channel.Drive.group.vibration3",
    "Channel.Drive.group.vibration5",
    "Channel.Drive.group.vibration2",
    "温度传感器5",
    "温度传感器6",
    "温度传感器7",
    "温度传感器8",
};

rd_kafka_conf_t *conf;
rd_kafka_conf_t *conf1;
rd_kafka_t *rk;
rd_kafka_t *rk1;
rd_kafka_topic_conf_t *topic_conf;
rd_kafka_topic_conf_t *topic_conf1;
rd_kafka_topic_t *rkt;
rd_kafka_topic_t *rkt1;

modbus_t *ctx;
char errstr[512];
int partition = RD_KAFKA_PARTITION_UA;

uint16_t tab_reg[SUM_SLAVE][8];
int rc;

int md_init(void)
{
    ctx = modbus_new_rtu("/dev/ttySAC2",9600,'N',8,1);

    if(NULL == ctx)
    {
        fprintf(stderr,"Unable to create the libmodbus context\n");
        return -1;
    }
    else
    {
        fprintf(stdout,"Create the libmodbus connect\n");
    }
    return 0;
}

int md_read(int slave)
{
    modbus_set_debug(ctx,slave);
    modbus_set_slave(ctx,slave);

    int i;
    if(-1 == modbus_connect(ctx))
    {
		fprintf(stderr,"Connect failed:%s\n",modbus_strerror(errno));
		modbus_free(ctx);
		return -1;
	}
	else
	{
		fprintf(stdout,"Connect \n");
	}
	rc = modbus_read_input_registers(ctx,0,8,tab_reg[slave-1]);
	if(-1 == rc)
	{
		fprintf(stderr,"%s\n",modbus_strerror(errno));
		return -1;
	}
	for (i=0; i < rc; i++)
        {
		printf("reg[%d]=%d (0x%X)\n", i, tab_reg[slave-1][i], tab_reg[slave-1][i]);
	}

	return 0;
}

//i is the number of xml's efective sensor

int find_item_id(int i,int count)
{
    count = 0;
    int j,k;
    xmlNodePtr p;
    for(j=0;j<i;j++)
    {
        for(p=node[j],node[j]=node[j]->children;node[j];node[j]=node[j]->next)
        {
            if(!xmlStrcasecmp(node[j]->name,BAD_CAST"item_id"))
            {
                for(k=0;k<SUM_SENSOR;k++)
                {
                    if(strcmp(xmlNodeGetContent(node[j]),sensor[k]))
                    {
                        continue;
                    } 
                    else   //匹配成功
                    {
                        printf("k = %d\n",k);  //k is the index of sensor
                        save_to_arr(p,count,k);
                        //printf("----%s\n",xmlNodeGetContent(node[j]));
                        count++;
                        break;
                    }
                }
                if(k>=SUM_SENSOR)  
                {
                    printf("this collect module not include %s sensor\n",xmlNodeGetContent(node[j]));
                    break;
                }
                node[j]=NULL;
                break;
            }        
            else if(!node[j]->next)
            {
                printf("not found this tree(item_id)\n");  //xml not exists node item_id
                break;
            }
        }
    }
    return count;
}

int save_to_arr(xmlNodePtr p,int count,int k)  // count is the num of node = 8
{
    char str[4];
    sprintf(str,"%d",k);
    sprintf(state_info[count][4],"%s",str);
    int len;
    //put head node into head[i][]
    if(xmlStrcasecmp(p->name,BAD_CAST"collect_point")==0)
    {
        name = xmlGetProp(p,"name");
        value = xmlGetProp(p,"id");
        //printf("name=%s\nvalue=%s\n",name,value);
        strcpy(head[count][0],value);
        strcpy(head[count][1],name);
    }
    //get this node's children that is to say dev basic info
    for(p=p->children,len=0;p;p=p->next,len++)
    {
        //get node's id attribute
        if(len>=0&&len<=2)
        {
             name = xmlGetProp(p,"id");
             value = xmlNodeGetContent(p);
             strcpy(dev_basic[count][2*len],name);
             strcpy(dev_basic[count][2*len+1],value);
      //       printf("start----\n%s\n%s\n----end\n",dev_basic[count][2*len],dev_basic[count][2*len+1]);
             continue;
        }
        //get state info 
        if(len>2&&len<=6)
        {
            value = xmlNodeGetContent(p);
            strcpy(state_info[count][len-3],value);
        //    printf("state_info = %s\n",state_info[count][len-3]);
        }
        if((p->next==NULL)&&(p->children==NULL))
        {
            p = NULL;
            break;
        }// parser warning
        else if((p->next==NULL)&&(p->children))
        {
            if(!(xmlStrcasecmp(p->name,BAD_CAST"warning_state")))
            {
                p = p->children;
            }
            
            while(p)
            {
                value = xmlNodeGetContent(p);
                strcpy(warning_info[count][len-7],value);
                //printf("warning_info = %s\n",warning_info[count][len-7]);
                len++;
                if(p->next)
                    p = p->next;
                else
                    break;
            }
        }
    }
    xmlFree(name);
    xmlFree(value);
}

int warning(float max,float min,int value,int f)
{
    bzero(buf_test,sizeof(buf_test));
    bzero(buf_test1,sizeof(buf_test1));
    bzero(buf_test2,sizeof(buf_test2));
   
    if(!(strcmp(warning_info[f][0],"None") == 0))
    {
        sprintf(buf_test,"None");
        sprintf(buf_test1,"None");
        sprintf(buf_test2,"False");
        return 0; 
    }

    if(value>max || value<min)
    {
        sprintf(buf_test,"%s",warning_info[f][0]);	
	sprintf(buf_test1,"%s",warning_info[f][1]);	
	sprintf(buf_test2,"%s",warning_info[f][3]);
    }
    else
    {
        sprintf(buf_test,"None");
        sprintf(buf_test1,"None");
        sprintf(buf_test2,"False");
    }
    return 0;
}

int main(int argc,char **argv)
{
    int i = 0;  //num of the head node
    int k = 0;
    int f = 0;
    int delay = atoi(argv[2]);

    const char *brokers = argv[3];
    const char *topic = argv[4];
    const char *topic1 = argv[5];
    char ntp_ip[32];
    
    struct timeval tv;
    gettimeofday(&tv,NULL);
    time_t time_s = {time(NULL)};
    struct tm ts = {0}; 
   
    bzero(ntp_ip,sizeof(char)*32);
    sprintf(ntp_ip,"ntpdate -d %s",argv[6]);
    printf("%s\n",ntp_ip);
    system(ntp_ip);
    
    printf("---------md_init_start------------\n");
    md_init();
    printf("---------md_init_end--------------\n");
    printf("%d\n",delay);

    conf = rd_kafka_conf_new();
    conf1 = rd_kafka_conf_new();
    printf("---------kafka config-------------\n");

    topic_conf = rd_kafka_topic_conf_new();
    topic_conf1 = rd_kafka_topic_conf_new();
    printf("---------topic config-------------\n");
    
    if(NULL == (rk = rd_kafka_new(RD_KAFKA_PRODUCER,conf,errstr,sizeof(errstr))))
    {
        fprintf(stderr,"%% Failed to create new producer:%s\n",errstr);
        return -1;
    }
    if(NULL == (rk1 = rd_kafka_new(RD_KAFKA_PRODUCER,conf1,errstr,sizeof(errstr))))
    {
        fprintf(stderr,"%% Failed to create new producer:%s\n",errstr);
        return -1;
    }
    
    //add brokers
    if(rd_kafka_brokers_add(rk,brokers) == 0)
    {
        fprintf(stderr,"%% No valid brokers specified\n");
        return -1;
    }
    else
    {
        printf("rd_kafka_brokers_add is ok\n");
    }
    if(rd_kafka_brokers_add(rk1,brokers) == 0)
    {
        fprintf(stderr,"%% No valid brokers specified\n");
        return -1;
    }
    else
    {
        printf("rd_kafka_brokers_add is ok\n");
    }
    printf("---------add brokers--------------\n");

    rkt1 = rd_kafka_topic_new(rk1,topic1,topic_conf1);
    printf("---------topic his ok-------------\n");
    rkt = rd_kafka_topic_new(rk,topic,topic_conf);
    printf("---------topic cur ok-------------\n");

    xmlDocPtr doc;
    xmlNodePtr root;
    xmlNodePtr tmp=NULL;
    xmlKeepBlanksDefault(0);
    
    // load file
    if(!(doc=xmlParseFile(argv[1])))
    {
        printf("loading xml file error\n");
        exit(1);
    }

    //get root node
    if(!(root=xmlDocGetRootElement(doc)))
    {
        printf("empty file\n");
        xmlFreeDoc(doc);
        exit(2);
    }
    for(node[i]=root->children;node[i],i<80;tmp=node[i]->next,i++,node[i]=tmp)
    {
        while(xmlStrcasecmp(node[i]->name,BAD_CAST"collect_point")!=0)
        {
            node[i]=node[i]->next;
            if(node[i]==NULL)
                break;
        }
        if(node[i]->next==NULL)
        {
            if(i==0)
            {
                xmlFreeDoc(doc);
                exit(3);
            }
            break;
            printf("load finish or type is wrong\n");
        }
        tmp=NULL;
    }
    k = find_item_id(i,count);
    int tmp_value; 

    while(1)
    {
        i = SUM_SENSOR;
        while(i>0)
        {
            md_read((i/9+1));
            i-=8;
            usleep(20000);
        }
        bzero(buf_cur,sizeof(char)*2048);
        bzero(buf_his,sizeof(char)*2048);
        for(f = 0; f < k; f++)
        {
            tmp_value = atoi(state_info[f][4]);
            time(&time_s);
            localtime_r(&time_s,&ts);
            float real_value = ((tab_reg[tmp_value/8][tmp_value%8]*0.004)-4)*31.25;
            warning(atof(state_info[f][1]),atof(state_info[f][0]),real_value,f);
            sprintf(buf_cur,"{\"asset_id\":\"%s\",\"oid\":\"%s\",\"attr_id\":\"%s\",\"attr_name\":\"%s\",\"value\":\"%.1f\",\"collect_time\":\"%d:%d:%d:%ld\",\"warning_state\":\"%s\",\"warning_name\":\"%s\"}",dev_basic[f][4],state_info[f][3],dev_basic[f][0],dev_basic[f][1],real_value,ts.tm_hour,ts.tm_min,ts.tm_sec,tv.tv_usec/1000,buf_test,buf_test1);
            printf("%s\n",buf_cur);
           
            if(rd_kafka_produce(rkt,partition,RD_KAFKA_MSG_F_COPY,buf_cur,sizeof(char)*strlen(buf_cur),NULL,0,NULL) == -1)
            {
                fprintf(stderr,"%% Failed to produce to topic %s " "partition %i: %s\n",rd_kafka_topic_name(rkt),partition,rd_kafka_err2str(rd_kafka_errno2err(errno)));
            }
            bzero(buf_cur,sizeof(char)*(2048));
            
            gettimeofday(&tv,NULL);
            time(&time_s);
            localtime_r(&time_s,&ts);
            sprintf(buf_his,"{\"collect_point_id\":\"%s\",\"collect_point_name\":\"%s\",\"attribute_id\":\"%s\",\"attribute_name\":\"%s\",\"asset_id\":\"%s\",\"asset_name\":\"%s\",\"control_asset_id\":\"%s\",\"control_asset_name\":\"%s\",\"upper_limit\":\"%s\",\"lower_limit\":\"%s\",\"item_id\":\"%s\",\"warning_state\":\"%s\",\"warning_name\":\"%s\",\"trigger_repair_so\":\"%s\",\"value\":%.1f,\"collect_time\":\"%d:%d:%d:%ld\"}",head[f][0],head[f][1],dev_basic[f][0],dev_basic[f][1],dev_basic[f][2],dev_basic[f][3],dev_basic[f][4],dev_basic[f][5],state_info[f][0],state_info[f][1],state_info[f][2],buf_test,buf_test1,buf_test2,real_value,ts.tm_hour,ts.tm_min,ts.tm_sec,tv.tv_usec/1000);
	    printf("%s\n",buf_his);
            
            if(rd_kafka_produce(rkt1,partition,RD_KAFKA_MSG_F_COPY,buf_his,sizeof(char)*strlen(buf_his),NULL,0,NULL) == -1)
            {
                fprintf(stderr,"%% Failed to produce to topic %s " "partition %i: %s\n",rd_kafka_topic_name(rkt),partition,rd_kafka_err2str(rd_kafka_errno2err(errno)));
            }
            bzero(buf_his,sizeof(char)*(2048));
            usleep(delay);
            modbus_close(ctx);
        }
    }
    
    modbus_free(ctx);
    return 0;
}
